package com.examples.demo;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import okhttp3.ResponseBody;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class BaiduHotTopery {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        SingleOutputStreamOperator<JSONArray> dfdgdg = env.addSource(new RichSourceFunction<JSONArray>() {
            private static final long serialVersionUID = 33342353984591675L;
            @Override
            public void run(SourceContext<JSONArray> sourceContext) throws Exception {
                while (true){
                    OkHttpClient client = new OkHttpClient().newBuilder()
                            .build();
                    Request request = new Request.Builder()
                            .url("https://top.baidu.com/board?tab=realtime")
                            .method("GET", null)
                            .addHeader("User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 Edg/120.0.0.0")
                            .build();
                    Response response = client.newCall(request).execute();
                    String string = response.body().string();

                    Pattern regex = Pattern.compile("\\<\\!\\-\\-s\\-data\\:(.*?)\\-\\-\\>");
                    Matcher matcher = regex.matcher(string);
                    while (matcher.find()){
                        try {
                            String group = matcher.group(1);
                            JSONObject t = JSON.parseObject(group);
                            JSONObject data = (JSONObject) t.get("data");
                            JSONArray jsonArray = (JSONArray) data.get("cards");
                            JSONObject jsonObject = (JSONObject) jsonArray.get(0);
                            JSONArray jsonArray1 = (JSONArray) jsonObject.get("content");
                            sourceContext.collect(jsonArray1);
                        }catch (Exception ignored){}
                    }

                    Thread.sleep(1000*60);
                }
            }

            @Override
            public void cancel() {
            }
        }).setParallelism(1).name("baiduHotTopSource");
        SingleOutputStreamOperator<JSONObject> baiduHotTopFlatMap = dfdgdg.flatMap(new FlatMapFunction<JSONArray, JSONObject>() {
            @Override
            public void flatMap(JSONArray objects, Collector<JSONObject> collector) throws Exception {
                objects.forEach(o -> {
                    collector.collect((JSONObject) o);
                });
            }
        }).setParallelism(5).name("baiduHotTopFlatMap");
        SingleOutputStreamOperator<Tuple2<String, JSONObject>> tuple2SingleOutputStreamOperator = baiduHotTopFlatMap.flatMap(new FlatMapFunction<JSONObject, Tuple2<String, JSONObject>>() {
            @Override
            public void flatMap(JSONObject jsonObject, Collector<Tuple2<String, JSONObject>> collector) throws Exception {
                collector.collect(new Tuple2<String, JSONObject>(String.valueOf(System.currentTimeMillis()), jsonObject));
            }
        }).name("TimeMillisTagData");
        OutputTag<Tuple3<String, String, String>> outputTag = new OutputTag<Tuple3<String, String, String>>("imageUrlTag", Types.TUPLE(Types.STRING, Types.STRING, Types.STRING));
        SingleOutputStreamOperator<Tuple2<String, JSONObject>> singleOutputStreamOperator = tuple2SingleOutputStreamOperator.process(new ProcessFunction<Tuple2<String, JSONObject>, Tuple2<String, JSONObject>>() {
            @Override
            public void processElement(Tuple2<String, JSONObject> stringJSONObjectTuple2, Context context, Collector<Tuple2<String, JSONObject>> collector) throws Exception {
                String[] imgs = stringJSONObjectTuple2.f1.getString("img").split("/");
                String img = imgs[imgs.length - 1];
                if(!img.contains("\\.")){
                    img = img + ".jpg";
                }
                String name = "D:\\ffffff\\2024-01-29--15\\" + img;
                context.output(outputTag, new Tuple3<String, String, String>(stringJSONObjectTuple2.f0, stringJSONObjectTuple2.f1.getString("img"), name));
                stringJSONObjectTuple2.f1.put("localImageFile", name);
                collector.collect(stringJSONObjectTuple2);
            }
        }).name("outputTag2dowImage");
        SideOutputDataStream<Tuple3<String, String, String>> sideOutput = singleOutputStreamOperator.getSideOutput(outputTag);
        sideOutput.addSink(new SinkFunction<Tuple3<String, String, String>>() {
            @Override
            public void invoke(Tuple3<String, String, String> value, Context context) throws Exception {
                SinkFunction.super.invoke(value, context);
                File file = new File(value.f2);
                if(!file.exists()){
                    OkHttpClient client = new OkHttpClient();
                    Request request = new Request.Builder()
                            .url(value.f1)
                            .build();
                    try (Response response = client.newCall(request).execute()) {
                        if (!response.isSuccessful()) {
                            throw new IOException("Unexpected code " + response);
                        }
                        ResponseBody body = response.body();
                        if (body != null) {
                            // 一次性读取所有字节
                            byte[] imageBytes = body.bytes();
                            BufferedOutputStream out = new BufferedOutputStream(new FileOutputStream(value.f2));
                            out.write(imageBytes);
                            out.close();
                        }
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        }).setParallelism(10).name("dowImage");
        SingleOutputStreamOperator<JSONObject> float2JSON = singleOutputStreamOperator.flatMap(new FlatMapFunction<Tuple2<String, JSONObject>, JSONObject>() {
            @Override
            public void flatMap(Tuple2<String, JSONObject> stringJSONObjectTuple2, Collector<JSONObject> collector) throws Exception {
                collector.collect(stringJSONObjectTuple2.f1);
            }
        }).setParallelism(2).name("float2JSON");

        float2JSON.addSink(new SinkFunction<JSONObject>() {
            @Override
            public void invoke(JSONObject value, Context context) throws Exception {
                SinkFunction.super.invoke(value, context);
                System.out.println("value = " + value);
            }
        }).name("over");




        env.execute("BaiduHotToper");


    }
}
